package com.coolpad.MQ.kafka;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.serializer.Encoder;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;


import java.util.Properties;

public class KafkaProducerWrapper <K, V> implements ProducerBase<K, V> {
    private static final Logger LOGGER = LoggerFactory.getLogger(com.coolpad.MQ.kafka.KafkaProducerWrapper.class);

    private Properties properties;

    private boolean started = false;
    private volatile boolean closed = false;
    private KafkaProducer<K, V> producer;

    @Override
    public void init(Properties config) {
        properties = config;
        //TODO Shutdown Hook
    }

    /**
     * kafka producer的encoder 直接可以在配置中指定，这里不再单独配置
     * @param keyEncoder key encoder
     * @return this
     */
    @Override
    public ProducerBase<K, V> setKeyEncoder(Encoder<K> keyEncoder) {
        return this;
    }

    /**
     * * kafka producer的encoder 直接可以在配置中指定，这里不再单独配置
     * @param contentEncoder content encoder
     * @return this
     */
    @Override
    public ProducerBase<K, V> setContentEncoder(Encoder<V> contentEncoder) {
        return this;
    }

    @Override
    public void start() {
        if (started) {
            throw new IllegalStateException("KafkaProducer.start can only be called once!");
        }

        if (!properties.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
            LOGGER.error("Cannot find kafka producer config of {} from {}!", ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    properties);
            System.exit(1);
        }
        // LOGGER.info("start kafka producer with config:{}", properties);
        producer = new KafkaProducer<>(properties);
        started = true;
    }

    @Override
    public boolean send(String topic, K key, V message) {
        if (closed) {
            LOGGER.error("KafkaProducer has already closed! Discard message {} {} {}", topic, key, message);
            return false;
        }
        LOGGER.debug("send kafka message: topic={} key={} message={}", topic, key, message);
        ProducerRecord<K, V> record = new ProducerRecord<>(topic, key, message);
        long start = System.currentTimeMillis();
        producer.send(record);
        return true;
    }

    @Override
    public void close() {
        if (closed) {
            return;
        }
        closed = true;
        LOGGER.info("Close Kafka producer ...");
        if (producer != null) {
            producer.close();
        }
    }

    @Override
    public String typeName() {
        return "kafka";
    }
}

